package com.qdu.hyy.MovieHighestRating;
import com.qdu.hyy.MovieWritable;
import com.qdu.hyy.PopularityCount.PopularityCountDriver;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class MovieHighestRatingDriver {
    public static void main(String[] args) throws Exception {
        // 确保提供了足够的参数
//        if (args.length != 2) {
//            System.err.println("Usage: TopNMoviesDriver <input path> <output path>");
//            System.exit(-1);
//        }

        // 创建配置对象
        Configuration conf = new Configuration();

        // 设置要找出评分最高的电影数量（可选项，这里硬编码为10）
        // 你也可以从命令行参数或配置文件中读取这个值
        conf.setInt("top.n.movies", 10);

        // 创建一个Job对象
        Job job = Job.getInstance(conf, "Top N Movies");

        // 设置Job的Driver类
        job.setJarByClass(PopularityCountDriver.class);

        // 设置Mapper和Reducer类
        job.setMapperClass(MovieHighestRatingMapper.class);
        job.setReducerClass(MovieHighestRatingReducer.class);

        // 设置Mapper输出键值对的类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(DoubleWritable.class);

        // 设置Reducer输出键值对的类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);

        // 设置输入和输出路径
        FileInputFormat.addInputPath(job, new Path("input/hyy/movies_complete.csv"));
        FileOutputFormat.setOutputPath(job, new Path("output/job2"));

        // 提交作业并等待其完成
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
